package cn.itcast.d_state;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.collection.mutable.HashMap;

import java.util.Arrays;

//需求: 使用MapState对本地集合进行求和 (根据第一个字段分组)
public class MapStateDemo2 {
    public static void main(String[] args) throws Exception {
        /**
         * 1.获取流处理执行环境
         * 2.加载本地集合
         * 3.数据分组
         * 4.数据求和
         * 5.数据打印
         * 6.触发执行
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<Long, Long>> source = env.fromCollection(Arrays.asList(
                Tuple2.of(1L, 4L),
                Tuple2.of(2L, 3L),
                Tuple2.of(3L, 1L),
                Tuple2.of(1L, 2L),
                Tuple2.of(3L, 2L),
                Tuple2.of(1L, 2L),
                Tuple2.of(2L, 2L),
                Tuple2.of(2L, 9L)));

        //1,8
        //2,14
        //3,3
        //3.数据分组
        //mapstate作为一个状态内存对象，缓存的是中间结果，供下一次使用
        source
                .keyBy(line -> line.f0)
//                        .keyBy(new KeySelector<Tuple2<Long, Long>, Long>() {
//                            @Override
//                            public Long getKey(Tuple2<Long, Long> value) throws Exception {
//                                return value.f0;
//                            }
//                        })


        .map(new RichMapFunction<Tuple2<Long, Long>, Object>() {
            //初始化方法，在主方法运行之前，优先执行，而且仅执行一遍
            MapState<Long, Long> mapState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                //设置mapstate的kv类型
                MapStateDescriptor<Long, Long> msStateDsc = new MapStateDescriptor<>("msState", Long.class, Long.class);
                //获取mapstate对象
                mapState = getRuntimeContext().getMapState(msStateDsc);
            }

            @Override
            public Object map(Tuple2<Long, Long> value) throws Exception {
                //获取mapstate的中间结果值
                Long valueLast = mapState.get(value.f0);
                //做非空判断
                valueLast = valueLast == null?0l:valueLast;

                //与最新数据进行累加
                Tuple2<Long, Long> of = Tuple2.of(value.f0, value.f1 + valueLast);

                //把最新中间结果，缓存到内存里-》mapState
                mapState.put(of.f0,of.f1);

                return of;
            }
        }).print();


        env.execute();
    }
}
